src/handlers/http/modal/ingest/ingestor_logstream.rs (1)

65-65: Prefer 204 No Content over 200 "Cleanup complete"

The response body isn’t consumed by callers; returning 204 reduces noise and aligns with "action completed, no payload".

-    Ok(("Cleanup complete", StatusCode::OK))
+    Ok(actix_web::HttpResponse::NoContent().finish())

src/prism/logstream/mod.rs (1)

155-155: Use PARSEABLE.storage() accessor for consistency (and to avoid depending on field visibility).

Prefer the public accessor to obtain the storage provider, matching the rest of the codebase.

-    let storage = PARSEABLE.storage.get_object_store();
+    let storage = PARSEABLE.storage().get_object_store();

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/parseable/mod.rs:528-533
Timestamp: 2025-08-18T12:37:47.732Z
Learning: In Parseable, the validate_time_partition function in src/utils/json/flatten.rs already provides a default time partition limit of 30 days using `map_or(30, |days| days.get() as i64)` when time_partition_limit is None, so no additional defaulting is needed in the stream creation logic in src/parseable/mod.rs.

Learnt from: de-sh
PR: parseablehq/parseable#1185
File: src/handlers/http/logstream.rs:255-261
Timestamp: 2025-02-14T09:49:25.818Z
Learning: In Parseable's logstream handlers, stream existence checks must be performed for both query and standalone modes. The pattern `!PARSEABLE.streams.contains(&stream_name) && (PARSEABLE.options.mode != Mode::Query || !PARSEABLE.create_stream_and_schema_from_storage(&stream_name).await?)` ensures proper error handling in both modes.

src/migration/mod.rs (1)

369-369: Standardize storage accessor usage for consistency

Switched to PARSEABLE.storage().get_object_store(), which is correct per the new accessor. Elsewhere in this file we still use config.storage.get_object_store(). Consider standardizing on the accessor to avoid confusion and keep a single pattern across modules.

src/handlers/http/modal/ingest/ingestor_logstream.rs (1)

41-52: Prefer using check_or_load_stream for mode-aware existence checks

This endpoint currently re-implements a existence check using streams.contains + create_stream_and_schema_from_storage. We already have a helper that handles mode differences and storage loading: PARSEABLE.check_or_load_stream(&stream_name). It improves readability and keeps logic consistent with other handlers.

Suggested change:

-    // if the stream not found in memory map,
-    //check if it exists in the storage
-    //create stream and schema from storage
-    if !PARSEABLE.streams.contains(&stream_name)
-        && !PARSEABLE
-            .create_stream_and_schema_from_storage(&stream_name)
-            .await
-            .unwrap_or(false)
-    {
-        return Err(StreamNotFound(stream_name.clone()).into());
-    }
+    // Ensure parseable is aware of stream in distributed mode
+    if !PARSEABLE.check_or_load_stream(&stream_name).await {
+        return Err(StreamNotFound(stream_name.clone()).into());
+    }

src/storage/mod.rs (1)

133-141: Only server-side serialization of StreamInfo detected – no deserialization occurs

A search for serde_json::from_*<StreamInfo> or web::Json<StreamInfo> returned no matches. That means the service only emits StreamInfo (now in camelCase) and never consumes it.

• Document this breaking change for any API consumers.
• If in the future you need to accept incoming snake_case payloads, consider adding serde aliases on the key fields in src/storage/mod.rs:

#[serde(alias = "created_at")]
pub created_at: String,
#[serde(alias = "first_event_at")]
pub first_event_at: Option<String>,
#[serde(alias = "latest_event_at")]
pub latest_event_at: Option<String>,

src/handlers/http/cluster/mod.rs (1)

599-632: Retaining cleanup API: signature and behavior changes look good; include status in error for better diagnostics

Returning Result<(), ObjectStorageError> and accepting dates: &[String] fits the new orchestration model. Minor improvement: include the HTTP status in logs/errors to aid debugging.

-    if !resp.status().is_success() {
-        let body = resp.text().await.unwrap_or_default();
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let body = resp.text().await.unwrap_or_default();
         error!(
-            "failed to perform cleanup on retention: {}\nResponse Returned: {:?}",
-            ingestor.domain_name, body
+            "failed to perform cleanup on retention: {} (status: {})\nResponse Returned: {:?}",
+            ingestor.domain_name, status, body
         );
-        return Err(ObjectStorageError::Custom(format!(
-            "failed to perform cleanup on retention: {}\nResponse Returned: {:?}",
-            ingestor.domain_name, body
-        )));
+        return Err(ObjectStorageError::Custom(format!(
+            "failed to perform cleanup on retention: {} (status: {})\nResponse Returned: {:?}",
+            ingestor.domain_name, status, body
+        )));
     }

src/handlers/http/logstream.rs (2)

336-351: Single-call fetch for first/latest timestamps: solid; consider caching to reduce storage reads

Using storage.get_first_and_latest_event_from_storage is a clear improvement and aligns with the new StreamInfo.latest_event_at. One consideration: this now hits object storage on every info request. If this endpoint is hot, consider caching these values on the stream metadata (best-effort) and/or adding lightweight memoization with short TTL to reduce storage I/O. Optional and can be a follow-up.

---

415-421: Consistent accessor usage here is good; consider applying uniformly across the module

This spot adopts PARSEABLE.storage().get_object_store(); elsewhere in this file we still have PARSEABLE.storage.get_object_store(). For readability and future refactors, consider standardizing on the accessor.

src/prism/logstream/mod.rs (2)

155-170: Good move: single storage hit for both timestamps with graceful degradation.

Fetching both first/latest event timestamps via the storage provider simplifies logic and avoids repeated calls. On failure, logging a warning and defaulting to (None, None) is acceptable for a best-effort field.

One small improvement: if storage errors are transient (e.g., eventual consistency), consider downgrading to debug for noisy environments or adding error kind to the log to help triage.

---

184-184: StreamInfo now includes latest_event_at — verify serialization contract.

Assuming StreamInfo is annotated with serde camelCase (latestEventAt), this maps cleanly to the API surface. Double-check cross-file consistency (HTTP handlers and any UI/clients) for the new field name.

src/storage/object_storage.rs (4)

885-915: Happy path reuse of a single manifest is good; add a quick short-circuit for empty manifests.

Current logic already handles empty manifest_list early. Consider short-circuiting when min/max manifest paths are identical (you already do). No further action required beyond the min/max selection fix.

---

925-957: Consider using existing get_manifest instead of reaching through DataFusion registry.

load_manifest_from_path goes via QUERY_SESSION and object_store_registry, which ties this helper to the query runtime and assumes the store is registered there with appropriate auth. You already have trait methods for fetching objects; if manifest_path can be translated to a RelativePath (or you can add a helper to do so), prefer using self.get_object(...) to keep concerns within ObjectStorage.

If absolute URLs are required, at least document the reliance on QUERY_SESSION registration and clarify thread-safety/perf expectations.

---

1068-1069: Return shape adjustment looks consistent with the upload pipeline.

process_parquet_files returning only manifest files simplifies the calling site. Make sure downstream snapshot update logic doesn’t rely on the removed status flag.

---

1084-1104: Concurrent upload path is OK; semaphore bound remains high but unchanged.

No issues; bound of 100 may be high for some backends — consider making it configurable if you see pressure in production.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1392
File: src/migration/stream_metadata_migration.rs:303-322
Timestamp: 2025-07-28T17:10:39.448Z
Learning: In Parseable's migration system (src/migration/stream_metadata_migration.rs), each migration function updates the metadata to the current latest format using CURRENT_OBJECT_STORE_VERSION and CURRENT_SCHEMA_VERSION constants, rather than producing incremental versions. For example, v5_v6 function produces v7 format output when these constants are set to "v7", not v6 format.

Learnt from: de-sh
PR: parseablehq/parseable#1185
File: src/handlers/http/logstream.rs:255-261
Timestamp: 2025-02-14T09:49:25.818Z
Learning: In Parseable's logstream handlers, stream existence checks must be performed for both query and standalone modes. The pattern `!PARSEABLE.streams.contains(&stream_name) && (PARSEABLE.options.mode != Mode::Query || !PARSEABLE.create_stream_and_schema_from_storage(&stream_name).await?)` ensures proper error handling in both modes.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/handlers/http/ingest.rs:163-164
Timestamp: 2025-08-18T19:10:11.941Z
Learning: Field statistics calculation in src/storage/field_stats.rs uses None for the time_partition parameter when calling flatten_and_push_logs(), as field stats generation does not require time partition functionality.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/parseable/mod.rs:528-533
Timestamp: 2025-08-18T12:37:47.732Z
Learning: In Parseable, the validate_time_partition function in src/utils/json/flatten.rs already provides a default time partition limit of 30 days using `map_or(30, |days| days.get() as i64)` when time_partition_limit is None, so no additional defaulting is needed in the stream creation logic in src/parseable/mod.rs.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1340
File: src/storage/object_storage.rs:832-843
Timestamp: 2025-06-18T06:45:37.070Z
Learning: Stats calculation for parquet files in Parseable is done synchronously during the upload process because files are deleted from staging after upload. This prevents race conditions and ensures stats are calculated while files are still available locally.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/storage/object_storage.rs:997-1040
Timestamp: 2025-08-18T14:56:18.463Z
Learning: In Parseable's staging upload system (src/storage/object_storage.rs), failed parquet file uploads should remain in the staging directory for retry in the next sync cycle, while successful uploads remove their staged files immediately. Early return on first error in collect_upload_results is correct behavior as concurrent tasks handle their own cleanup and failed files need to stay for retry.

src/storage/object_storage.rs (1)

975-1019: Timestamp extraction correctly skips invalid values; previous epoch-default bug resolved.

Good job avoiding unwrap_or_default and only considering valid conversions/parses for min/max. This prevents skewed epoch results.

src/storage/object_storage.rs (3)

877-889: Avoid Option::is_none_or to prevent MSRV surprises; use map_or instead.

is_none_or is relatively new and can cause MSRV bumps. map_or reads fine and is widely supported.

-        for &item in &all_manifest_items {
-            if first_manifest_item
-                .is_none_or(|cur: &ManifestItem| item.time_lower_bound < cur.time_lower_bound)
-            {
-                first_manifest_item = Some(item);
-            }
-            if latest_manifest_item
-                .is_none_or(|cur: &ManifestItem| item.time_upper_bound > cur.time_upper_bound)
-            {
-                latest_manifest_item = Some(item);
-            }
-        }
+        for &item in &all_manifest_items {
+            if first_manifest_item
+                .map_or(true, |cur: &ManifestItem| item.time_lower_bound < cur.time_lower_bound)
+            {
+                first_manifest_item = Some(item);
+            }
+            if latest_manifest_item
+                .map_or(true, |cur: &ManifestItem| item.time_upper_bound > cur.time_upper_bound)
+            {
+                latest_manifest_item = Some(item);
+            }
+        }

---

932-964: Make manifest path handling robust for absolute URLs (s3://, gs://) vs relative keys.

object_store::path::Path::parse typically expects a key relative to the store, not a full URL. If manifest_path is an absolute URL (it is created via absolute_url during upload), parsing directly may fail on some backends. Normalize the path by stripping the URL and passing only the key to get().

-        let path = object_store::path::Path::parse(manifest_path)
-            .map_err(|e| ObjectStorageError::UnhandledError(Box::new(e)))?;
+        // Support both absolute URLs (e.g., s3://bucket/prefix/file) and relative keys (prefix/file)
+        let path = match url::Url::parse(manifest_path) {
+            Ok(u) => {
+                // Use only the key part for the store
+                let key = u.path().trim_start_matches('/');
+                object_store::path::Path::parse(key)
+                    .map_err(|e| ObjectStorageError::UnhandledError(Box::new(e)))?
+            }
+            Err(_) => object_store::path::Path::parse(manifest_path)
+                .map_err(|e| ObjectStorageError::UnhandledError(Box::new(e)))?,
+        };

---

1142-1181: Unreachable branch in collect_upload_results; simplify and clarify cleanup semantics.

UploadResult.manifest_file is always Some on success and errors return Err, so the None branch is dead. Removing it clarifies intent and avoids confusion.

-            Ok(Ok(upload_result)) => {
-                if let Some(manifest_file) = upload_result.manifest_file {
-                    uploaded_files.push((upload_result.file_path, manifest_file));
-                } else {
-                    // File failed to upload, clean up
-                    if let Err(e) = remove_file(upload_result.file_path) {
-                        warn!("Failed to remove staged file: {e}");
-                    }
-                }
-            }
+            Ok(Ok(upload_result)) => {
+                // Success: record manifest and remove staged file later
+                if let Some(manifest_file) = upload_result.manifest_file {
+                    uploaded_files.push((upload_result.file_path, manifest_file));
+                }
+            }

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/handlers/http/ingest.rs:163-164
Timestamp: 2025-08-18T19:10:11.941Z
Learning: Field statistics calculation in src/storage/field_stats.rs uses None for the time_partition parameter when calling flatten_and_push_logs(), as field stats generation does not require time partition functionality.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/parseable/mod.rs:528-533
Timestamp: 2025-08-18T12:37:47.732Z
Learning: In Parseable, the validate_time_partition function in src/utils/json/flatten.rs already provides a default time partition limit of 30 days using `map_or(30, |days| days.get() as i64)` when time_partition_limit is None, so no additional defaulting is needed in the stream creation logic in src/parseable/mod.rs.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1340
File: src/storage/object_storage.rs:832-843
Timestamp: 2025-06-18T06:45:37.070Z
Learning: Stats calculation for parquet files in Parseable is done synchronously during the upload process because files are deleted from staging after upload. This prevents race conditions and ensures stats are calculated while files are still available locally.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/storage/object_storage.rs:997-1040
Timestamp: 2025-08-18T14:56:18.463Z
Learning: In Parseable's staging upload system (src/storage/object_storage.rs), failed parquet file uploads should remain in the staging directory for retry in the next sync cycle, while successful uploads remove their staged files immediately. Early return on first error in collect_upload_results is correct behavior as concurrent tasks handle their own cleanup and failed files need to stay for retry.

src/prism/home/mod.rs (4)

43-51: Prefer a named struct over a 4-tuple for readability and future evolution

The tuple has grown to four elements and is getting unwieldy. A dedicated struct (e.g., StreamMetadataAggregate) will make call sites self-documenting and reduce ordering mistakes when adding/removing fields.

If you want to keep it lightweight now, consider at least using the alias consistently at call sites (see suggestion below for get_stream_metadata signature).

---

61-67: Align JSON casing with other APIs by using camelCase

Nice addition. To keep the Home API consistent with the rest (StreamInfo moved to camelCase), consider emitting camelCase here as well.

Apply:

 #[derive(Debug, Serialize)]
-pub struct DataSet {
+#[serde(rename_all = "camelCase")]
+pub struct DataSet {
   title: String,
   dataset_type: TelemetryType,
   #[serde(skip_serializing_if = "Option::is_none")]
   time_partition: Option<String>,
 }

---

215-225: Use the StreamMetadataResponse alias in the signature for maintainability

You already introduced a type alias; using it here will keep the function signature concise and future-proof if the payload changes again.

-async fn get_stream_metadata(
-    stream: String,
-) -> Result<
-    (
-        String,
-        Vec<ObjectStoreFormat>,
-        TelemetryType,
-        Option<String>,
-    ),
-    PrismHomeError,
-> {
+async fn get_stream_metadata(stream: String) -> StreamMetadataResponse {

---

255-257: Don’t assume the first manifest carries time_partition; pick the first non-empty across manifests

In mixed or evolving deployments, earlier manifests might not have time_partition populated while later ones do. Prefer the first non-empty value to avoid incorrectly returning None.

-    let time_partition = stream_jsons[0].time_partition.clone();
-
-    Ok((stream, stream_jsons, dataset_type, time_partition))
+    let time_partition = stream_jsons
+        .iter()
+        .find_map(|m| m.time_partition.clone());
+
+    Ok((stream, stream_jsons, dataset_type, time_partition))

Optionally, you might also sanity-check that telemetry_type is consistent across manifests and log a warn if it diverges.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

src/prism/home/mod.rs (4)

134-149: Using first stream.json for dataset_type/time_partition — consider consistency check across files

If multiple stream.json files exist and can diverge, silently taking index 0 may hide inconsistencies. Optional: assert they’re uniform and log if not, while still proceeding.

Apply within this block:

-                stream_wise_stream_json.insert(stream.clone(), metadata);
-                datasets.push(DataSet {
-                    title: stream,
-                    dataset_type,
-                    time_partition,
-                });
+                // Optional: sanity-check uniformity across metadata entries
+                let mixed_types = metadata.iter().any(|m| m.telemetry_type != dataset_type);
+                let mixed_partitions = metadata.iter().any(|m| m.time_partition != time_partition);
+                if mixed_types || mixed_partitions {
+                    tracing::warn!(
+                        stream = %stream,
+                        mixed_types,
+                        mixed_partitions,
+                        "Non-uniform stream metadata detected; using the first entry"
+                    );
+                }
+                stream_wise_stream_json.insert(stream.clone(), metadata);
+                datasets.push(DataSet {
+                    title: stream,
+                    dataset_type,
+                    time_partition,
+                });

Add at top of the file if needed:

use tracing::warn;

---

257-260: time_partition derived from first metadata entry — align with optional uniformity check

This follows the “first entry” approach; if you adopt the uniformity check above, this remains fine. Otherwise, be aware of potential divergence across entries.

---

160-165: Avoid cloning the entire stream metadata map per date (perf nit)

Cloning the full HashMap for each day multiplies memory traffic. Pass by reference instead; join_all supports non-'static futures with borrows.

Use these diffs:

-    let stats_futures = dates
-        .iter()
-        .map(|date| stats_for_date(date.clone(), stream_wise_stream_json.clone()));
-    let stats_results: Vec<Result<DatedStats, PrismHomeError>> =
-        futures::future::join_all(stats_futures).await;
+    let stats_futures = dates
+        .iter()
+        .map(|date| stats_for_date(date.clone(), &stream_wise_stream_json));
+    let stats_results: Vec<Result<DatedStats, PrismHomeError>> =
+        futures::future::join_all(stats_futures).await;

And update the callee signature accordingly (see below).

---

262-265: Match signature to accept a reference to avoid clones

Change stats_for_date to borrow the map.

-async fn stats_for_date(
-    date: String,
-    stream_wise_meta: HashMap<String, Vec<ObjectStoreFormat>>,
-) -> Result<DatedStats, PrismHomeError> {
+async fn stats_for_date(
+    date: String,
+    stream_wise_meta: &HashMap<String, Vec<ObjectStoreFormat>>,
+) -> Result<DatedStats, PrismHomeError> {

The function body remains the same; iteration over stream_wise_meta.values() continues to work.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/parseable/mod.rs:528-533
Timestamp: 2025-08-18T12:37:47.732Z
Learning: In Parseable, the validate_time_partition function in src/utils/json/flatten.rs already provides a default time partition limit of 30 days using `map_or(30, |days| days.get() as i64)` when time_partition_limit is None, so no additional defaulting is needed in the stream creation logic in src/parseable/mod.rs.

src/storage/object_storage.rs (2)

965-1020: Timestamp extraction handles invalid values correctly (thanks for fixing earlier unwrap_or_default issue)

• Int stats: guarded by from_timestamp_millis returning Some.
• String stats: guarded by RFC3339 parse; converted to Utc.

This aligns with the prior guidance to skip invalid values rather than default to epoch. Good job.

---

1142-1181: Do not delete staged files when no manifest is produced

Current branch removes the staged file if manifest_file is None. This violates the intended retry behavior for failures where we should keep files in staging. Also, "success-without-manifest" is suspicious; better to log and retain for retry rather than delete.

Apply this diff:

-                if let Some(manifest_file) = upload_result.manifest_file {
-                    uploaded_files.push((upload_result.file_path, manifest_file));
-                } else {
-                    // File failed to upload, clean up
-                    if let Err(e) = remove_file(upload_result.file_path) {
-                        warn!("Failed to remove staged file: {e}");
-                    }
-                }
+                if let Some(manifest_file) = upload_result.manifest_file {
+                    uploaded_files.push((upload_result.file_path, manifest_file));
+                } else {
+                    // No manifest produced; keep for retry
+                    warn!(
+                        "Upload yielded no manifest; keeping staged file for retry"
+                    );
+                }

Early return on first error is retained (good) and aligns with previous guidance that failed files should remain in staging for retry.

src/storage/object_storage.rs (3)

168-185: Prefer elevating log level for failed stats from trace to debug

Silent failures at trace can be missed in prod. Recommend debug for visibility without being noisy.

Apply this diff:

-        if let Err(err) =
-            calculate_field_stats(stream_name, path, schema, max_field_statistics).await
-        {
-            tracing::trace!(
+        if let Err(err) =
+            calculate_field_stats(stream_name, path, schema, max_field_statistics).await
+        {
+            tracing::debug!(
                 "Error calculating field stats for stream {}: {}",
                 stream_name,
                 err
             );
         }

---

932-964: Minor import cleanup in helper

Manifest is already imported at the top of the module. The local import duplicates it unnecessarily; retain only QUERY_SESSION to avoid redundant imports.

Apply this diff:

-        use crate::{catalog::manifest::Manifest, query::QUERY_SESSION};
+        use crate::query::QUERY_SESSION;

---

1170-1181: Parallel cleanup with rayon is acceptable, but optional

Using rayon for removing files is fine; the IO might dwarf CPU anyway. If you want to avoid engaging a separate threadpool, a simple sequential loop would also suffice. No action needed.

src/event/format/json.rs (1)

41-44: Add a brief Rustdoc to clarify p_timestamp semantics

Minor nit to help future readers: document that p_timestamp is ingestion/processing time, not the event's source timestamp.

 impl Event {
-    pub fn new(json: Value, p_timestamp: DateTime<Utc>) -> Self {
+    /// Create a JSON event with explicit Parseable ingestion time (`p_timestamp`).
+    ///
+    /// Note: `p_timestamp` represents the time at ingestion into Parseable (server/ingestor time).
+    /// The event's source time (if any) is still derived from the payload when a time partition is configured.
+    pub fn new(json: Value, p_timestamp: DateTime<Utc>) -> Self {
         Self { json, p_timestamp }
     }

src/utils/json/mod.rs (1)

78-115: Making apply_generic_flattening_for_partition public is appropriate

This enables reusing the same flattening semantics outside this module (e.g., pstats emission), keeping behavior consistent with the ingestion path.

Consider adding a short doc comment clarifying:

• It may return 1..N flattened objects.
• Validation is always enabled (true).
• It forwards time_partition_limit as-is (no defaults here). Per earlier learnings, defaults are handled in flatten/validate_time_partition, so nothing else is needed here.

src/storage/field_stats.rs (2)

95-101: Be careful failing hard on parquet-path datetime extraction

Currently, any mismatch in path format aborts stats computation. If deployment paths vary (common across filesystems and object stores), this will be fragile.

Recommend either:

• downgrade to a warning and fall back to Utc::now() (still marking the event with ingestion time), or
• return a sentinel timestamp and proceed.

If strictness is desired, ensure the extractor handles both dotted and hierarchical path formats; see suggested implementation change below.

---

146-166: Avoid schema fetch inside the loop and remove unwrap on size

Two nit-level improvements:

• Fetch the stats stream schema once before iterating to avoid redundant lookups.
• Replace unwrap() on serialization with proper error propagation to prevent panics on unexpected serde failures.

-    for json in vec_json {
-        let origin_size = serde_json::to_vec(&json).unwrap().len() as u64; // string length need not be the same as byte length
-        let schema = PARSEABLE
-            .get_stream(DATASET_STATS_STREAM_NAME)?
-            .get_schema_raw();
+    // Fetch schema once to avoid repeated lookups
+    let schema = PARSEABLE
+        .get_stream(DATASET_STATS_STREAM_NAME)?
+        .get_schema_raw();
+
+    for json in vec_json {
+        // Compute byte length safely; propagate serialization errors
+        let origin_size = serde_json::to_vec(&json)
+            .map_err(|e| PostError::Invalid(e.into()))?
+            .len() as u64; // byte length

src/connectors/kafka/processor.rs (1)

84-94: Attach ingestion time at creation — aligned with the new constructor

Supplying Utc::now() is consistent with the ingestion-time semantics for p_timestamp. Callers with access to upstream timestamps can optionally pass those instead.

If ConsumerRecord exposes a message timestamp, consider preferring it over Utc::now() for better fidelity (fall back to Utc::now() when absent).

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/handlers/http/ingest.rs:163-164
Timestamp: 2025-08-18T19:10:11.941Z
Learning: Field statistics calculation in src/storage/field_stats.rs uses None for the time_partition parameter when calling flatten_and_push_logs(), as field stats generation does not require time partition functionality.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1340
File: src/storage/object_storage.rs:31-41
Timestamp: 2025-06-18T11:15:10.836Z
Learning: DataFusion's parquet reader defaults to using view types (Utf8View, BinaryView) when reading parquet files via the schema_force_view_types configuration (default: true). This means StringViewArray and BinaryViewArray downcasting is required when processing Arrow arrays from DataFusion parquet operations, even though these types are behind nightly feature flags.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/parseable/mod.rs:528-533
Timestamp: 2025-08-18T12:37:47.732Z
Learning: In Parseable, the validate_time_partition function in src/utils/json/flatten.rs already provides a default time partition limit of 30 days using `map_or(30, |days| days.get() as i64)` when time_partition_limit is None, so no additional defaulting is needed in the stream creation logic in src/parseable/mod.rs.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1340
File: src/storage/object_storage.rs:832-843
Timestamp: 2025-06-18T06:45:37.070Z
Learning: Stats calculation for parquet files in Parseable is done synchronously during the upload process because files are deleted from staging after upload. This prevents race conditions and ensures stats are calculated while files are still available locally.

Learnt from: nikhilsinhaparseable
PR: parseablehq/parseable#1405
File: src/storage/object_storage.rs:997-1040
Timestamp: 2025-08-18T14:56:18.463Z
Learning: In Parseable's staging upload system (src/storage/object_storage.rs), failed parquet file uploads should remain in the staging directory for retry in the next sync cycle, while successful uploads remove their staged files immediately. Early return on first error in collect_upload_results is correct behavior as concurrent tasks handle their own cleanup and failed files need to stay for retry.